package com.yixun.service;

import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.yixun.component.RedisService;
import com.yixun.constant.SystemConstant;
import com.yixun.model.BreakdownInfo;
import com.yixun.model.SiteIntroduction;
import com.yixun.utils.SiteUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Service
@Slf4j
public class ContinuousNullService {

    @Autowired
    private RedisService redisService;

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    private static Gson gson = new Gson();

    /**
     * @Desc: 故障判断，判断站点是否有通道连续20条数据都为空
     **/
    @Async("asyncPool")
    public void continuousNull(List<SiteIntroduction> siteIntroductions){
        log.info("处理所有运营中的站点,判断通道是否持续（20条数据）为空故障...");
        long l = System.currentTimeMillis();
        //过滤不在运营中的站点
        siteIntroductions = siteIntroductions.stream()
                .filter(siteIntroduction -> siteIntroduction.getStatus().equals("run"))
                .collect(Collectors.toList());

        siteIntroductions.forEach(siteIntroduction -> {
            //站点存在采集数据，并且数据大于20条
            String siteNo = siteIntroduction.getSiteNo();
            String key = String.format(SystemConstant.SITE_DATA_KEY,siteNo);
            Boolean exists = redisService.exists(key);
            if(exists && redisService.llen(key)>=20){
                //拉取20条数据
                List<String> jsons = redisService.lrange(key, 0, 19);
                doSiteBreakdown(siteNo,jsons);
            }
        });
        log.info("处理耗时:" + (System.currentTimeMillis() - l) + "ms");
    }

    /**
     * @Desc: 判断站点是否存在故障，处理故障
     **/
    private void doSiteBreakdown(String siteNo, List<String> jsons){
        //转格式
        Type type = new TypeToken<List<String>>() {}.getType();
        List<List<String>> siteDatas = new ArrayList<>();//存放该站点的20条数据
        jsons.forEach(json->{
            siteDatas.add(gson.fromJson(json, type));
        });

        //站点通道数
        Integer ailseType = (siteDatas.get(0).size()-2)/20;

        for(int aisle=1;aisle<=ailseType;aisle++){

            Boolean assertNull = assertAisleBreakdown(siteDatas, aisle);
            String key = String.format(SystemConstant.BREAKDOWN_CONTINUOUS_NULL_KEY,siteNo,aisle);// "breakdown:%s_%s_CONTINUOUS_NULL"
            if(assertNull){//某个通道正在持续空故障
                Date currentDate = new Date();
                //报站点通道故障
                //故障已存在
                    //更新故障
                //故障不存在
                    //新增故障
                BreakdownInfo breakdownInfo = null;
                if(redisService.exists(key)){
                    String breakdownJson = redisService.get(key);
                    breakdownInfo = JSONObject.parseObject(breakdownJson, BreakdownInfo.class);
                    breakdownInfo.setStatus(BreakdownInfo.STATUS_WARNING);
                    breakdownInfo.setTime(String.valueOf(currentDate.getTime()));
                }else{
                    breakdownInfo = new BreakdownInfo();
                    //breakdown_uid
                    breakdownInfo.setUid(String.valueOf(redisService.incr(SystemConstant.BREAKDOWN_UID)));
                    breakdownInfo.setTime(String.valueOf(currentDate.getTime()));
                    breakdownInfo.setStatus(BreakdownInfo.STATUS_WARNING);
                    breakdownInfo.setAisleNo(String.valueOf(aisle));
                    breakdownInfo.setType(BreakdownInfo.TYPE_CONTINUOUS_NULL);
                    breakdownInfo.setSiteNo(siteNo);
                }
                //更新redis
                String breakdownInfoJson = JSONObject.toJSONString(breakdownInfo);
                redisService.set(key, breakdownInfoJson);//// "breakdown:%s_%s_CONTINUOUS_NULL"
                //推送至kafka
                kafkaTemplate.send(SystemConstant.BREAKDOWN_TOPIC,breakdownInfoJson);//breakdown_data
            }else{
                //存在故障
                    //关闭故障
                //不存在故障
                    //不处理
                if(redisService.exists(key)){
                    String breakdownJson = redisService.get(key);
                    BreakdownInfo breakdownInfo = JSONObject.parseObject(breakdownJson, BreakdownInfo.class);
                    breakdownInfo.setStatus(BreakdownInfo.STATUS_CLOSED);
                    breakdownInfo.setTime(String.valueOf(new Date().getTime()));
                    //删除redis
                    redisService.del(key);
                    //推送至kafka
                    kafkaTemplate.send(SystemConstant.BREAKDOWN_TOPIC,JSONObject.toJSONString(breakdownInfo));
                }
            }
        }

    }

    /**
     * @Desc: 判断某个通道是否持续空故障
     **/
    private Boolean assertAisleBreakdown(List<List<String>> siteDatas,Integer aisleNo){
        Integer startIndex = (aisleNo*20)-18;//通道的下标
        Integer endIndex = (aisleNo*20)+2;
        for (List<String> siteData : siteDatas) {
            long count = siteData.subList(startIndex, endIndex).stream().filter(StringUtils::isEmpty).count();
            if(count!=20){
                return false;
            }
        }
        return true;
    }

}
